package com.ferry.taskflow.taskFlow.tx;


import com.ferry.taskflow.taskFlow.base.IdAble;

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

/**
 * Created by daiyong
 * 不能使用线程池，需要和任务开启相同数量的线程数量
 */
@Slf4j
public class TxFlow extends IdAble {

	private List<TxStage> txStages = Lists.newArrayList();

	/**
	 * 原始事务管理
	 */
	private DataSourceTransactionManager dataSourceTransactionManager;

	/**
	 * 自定义事务管理器
	 */
	private TxCoordinater txCoordinater = new TxCoordinater(this);

	private ThreadPerTaskExecutor executor = new ThreadPerTaskExecutor();

	private TxFlow() {
		super();
	}

	public static TxFlow create() {
		return new TxFlow();
	}

	public TxFlow dataSourceTransactionManager(DataSourceTransactionManager dataSourceTransactionManager) {
		this.dataSourceTransactionManager = dataSourceTransactionManager;
		return this;
	}

	public DataSourceTransactionManager dataSourceTransactionManager() {
		return this.dataSourceTransactionManager;
	}
	public TxCoordinater txCoordinater() {
		return this.txCoordinater;
	}

	public TxFlow addTxStage(TxStage txStage) {
		txStage.txFlow(this);
		txStages.add(txStage);
		txCoordinater.expect(txStage.executors().size());
		return this;
	}

	public boolean start() {

		log.info("TxFlow start, flowID:{}", id());

		Assert.notEmpty(txStages, "flow阶段不能为空");

		boolean result = true;

		txCoordinater.startWatcher();

		//顺序执行stage
		for (int i = 0; i < txStages.size(); i++) {

			TxStage stage = txStages.get(i);

			List<TxTaskExecutor> executors = stage.executors();

			//不能使用线程池
			CompletableFuture[] futures = new CompletableFuture[executors.size()];
			for (int j = 0; j < executors.size(); j++) {
				TxTaskExecutor txTaskExecutor = executors.get(j);
				futures[j] = CompletableFuture.runAsync(() -> txTaskExecutor.execute(), executor);  //每个任务一个线程
			}

			//需要全部执行完毕，才能进行下一阶段
			try {
				CompletableFuture.allOf(futures).get();
			} catch (Exception e) {
				log.error("start exception", e);
				/**
				 * 理论上逻辑不会到这里，线程中有异常捕捉
				 * 不需要做特殊处理，只要提交或者回滚就可以，事务管理器会做事务的处理
				 */
				result = false;
			}
		}

		return result;
	}

	public void shutdown() {
		if (null != executor) {
			executor = null;
		}
		if (null != txCoordinater) {
			txCoordinater.shutdown();
		}
		log.info("TxFlow shutdown, flowID:{}", id());
	}

	static final class ThreadPerTaskExecutor implements Executor {
		public void execute(Runnable r) { new Thread(r).start(); }
	}
}
